package b.acct.test

import java.nio.charset.Charset

import akka.actor.ActorSystem
import akka.event.slf4j.SLF4JLogging
import akka.http.scaladsl.model.headers.RawHeader
import akka.stream.{ActorMaterializer, BidiShape}
import akka.stream.scaladsl._
import akka.util.ByteString
import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport
import org.squbs.pipeline.{Context, PipelineFlow, PipelineFlowFactory, RequestContext}
import spray.json._

import scala.concurrent.ExecutionContext.Implicits.global
/**
  * Created by lenovo on 2016/12/31.
  */
case class OneHTTPResult(resultcode: Int, reason: String)

trait OneJsonSupport extends SprayJsonSupport with DefaultJsonProtocol {
  implicit val f1 = jsonFormat2(OneHTTPResult)
}
class OneFlow extends PipelineFlowFactory with SLF4JLogging with OneJsonSupport {

  log.debug("OneFlow ...")
  override def create(context: Context)(implicit system: ActorSystem): PipelineFlow = {
    implicit val materializer = ActorMaterializer()
    BidiFlow.fromGraph(GraphDSL.create() { implicit b =>
      val inbound = b.add(Flow[RequestContext].map { rc =>
        log.debug(s"inbound: ${rc.httpPipeliningOrder}")
        rc.addRequestHeader(RawHeader("DummyRequest", "ReqValue"))
      })
      val outbound = b.add(Flow[RequestContext].map{ rc =>
        log.info("outbound")
//        这么做会把“流”数据给“物化”，再往后传递，就没法“物化”了。
//        rc.response.foreach { f =>
//          f.foreach { f1 =>
//            f1.entity.dataBytes.runFold(ByteString(""))(_ ++ _).map { data =>
//              val rs = data.decodeString(Charset.defaultCharset())
//              val json = rs.parseJson
//              //val rslt = json.convertTo[OneHTTPResult]
//              println(json)
//            }
//          }
//        }
        rc.addResponseHeader(RawHeader("DummyResponse", "ResValue"))
      })
      BidiShape.fromFlows(inbound, outbound)
    })
  }
}